package com.biubiushop.socket.nio;

import cn.hutool.log.Log;
import cn.hutool.log.LogFactory;
import com.biubiushop.repository.IEquipmentRepository;
import com.biubiushop.repository.IOrdersRepository;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.Channel;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.nio.charset.Charset;
import java.util.Iterator;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @author chenjia
 * @date 2019/7/16
 * 服务端
 **/
@Component
public class SocketChannelServer implements ApplicationRunner {

    private static final Log log = LogFactory.get();

    private static final String CODING_TYPE = "gbk";

    private Selector selector = null;
    private ServerSocketChannel serverSocketChannel = null;

    @Resource
    ThreadPoolTaskExecutor threadPoolTaskExecutor;

    @Resource
    IEquipmentRepository equipmentRepository;

    @Resource
    IOrdersRepository ordersRepository;

    @Resource
    StringRedisTemplate stringRedisTemplate;

    public void start() throws IOException {
        try {
            //创建selector
            selector = Selector.open();
            //创建channel通道
            serverSocketChannel = ServerSocketChannel.open();
            //设置为非阻塞模式
            serverSocketChannel.configureBlocking(false);
            //绑定监听端口
            serverSocketChannel.socket().bind(new InetSocketAddress(8888));
            //将channel注册到selector上
            serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
            //设备的socketId为key,scoketChannel为value
            ConcurrentHashMap<String, SocketChannel> socketChannelMap = new ConcurrentHashMap<>(16);

            while (true) {
                //获取channel数量
                int readyChannels = selector.select();
                if (readyChannels == 0) {
                    continue;
                }
                //获取可用channel集合
                Set<SelectionKey> selectionKeys = selector.selectedKeys();
                log.debug("====>>>>服务端轮询后需处理的连接数量为:" + selectionKeys.size());

                Iterator<SelectionKey> iterator = selectionKeys.iterator();

                while (iterator.hasNext()) {
                    SelectionKey selectionKey = iterator.next();
                    iterator.remove();

                    //根据就绪状态,调用对应方法处理业务逻辑

                    //接入事件
                    if (selectionKey.isAcceptable()) {
                        log.info("======>>>>>>客户端接入服务端");
                        acceptHandler(serverSocketChannel, selector);
                    }
                    //可读事件
                    if (selectionKey.isValid() && selectionKey.isReadable()) {
                        log.info("======>>>>>>客户端与服务端进行读写操作");
                        readHandler(selectionKey, socketChannelMap);
                    }
                }
            }
        } catch (IOException e) {
            e.printStackTrace();
            throw e;
        }
    }

    /**
     * socket接入处理
     *
     * @param serverSocketChannel
     * @param selector
     */
    private void acceptHandler(ServerSocketChannel serverSocketChannel, Selector selector) {
        SocketChannel socketChannel = null;
        try {
            //如果是接入事件,创建socketChannel
            socketChannel = serverSocketChannel.accept();

            //设置为非阻塞模式
            socketChannel.configureBlocking(false);

            //将channel注册到selector,监听可读事件
            socketChannel.register(selector, SelectionKey.OP_READ);

            //回复客户端
            socketChannel.write(Charset.forName(CODING_TYPE).encode("Hello, WELCOME TO SERVER!"));
        } catch (IOException e) {
            log.error("==============客户端接入异常==============");
            e.printStackTrace();
            try {
                socketChannel.close();
                log.error("==============关闭异常客户端连接==============");
            } catch (IOException ex) {
                ex.printStackTrace();
            }
        }

    }


    /**
     * @param selectionKey
     */
    private void readHandler(SelectionKey selectionKey, ConcurrentHashMap<String, SocketChannel> socketChannelMap) {
        SocketChannel socketChannel = null;
        ByteBuffer byteBuffer;
        ByteArrayOutputStream byteArrayOutputStream;
        try {
            //从selectionKey中获取已就绪的channel
            socketChannel = (SocketChannel) selectionKey.channel();

            //创建buffer
            byteBuffer = ByteBuffer.allocate(1024);
            byteArrayOutputStream = new ByteArrayOutputStream();

            //循环读取客户端请求信息
            String request;
            int read = socketChannel.read(byteBuffer);
            byte[] arr;

            while (read > 0) {
                arr = (byte[]) byteBuffer.flip().array();
                byteArrayOutputStream.write(arr, 0, read);
                if (read < byteBuffer.capacity()) {
                    request = byteArrayOutputStream.toString(CODING_TYPE);

                    if (request.trim().length() > 0) {
                        threadPoolTaskExecutor.execute(new ReadAndWriteHandlerThread(selectionKey, request, socketChannelMap, equipmentRepository, ordersRepository, stringRedisTemplate));
                    }
                    break;
                }
                read = socketChannel.read(byteBuffer);
            }
            if (read < 0) {
                log.info("===>>>客户端断开连接<<<===");
                socketChannel.close();
                selectionKey.cancel();
            }
        } catch (IOException e) {
            log.info("======>>>>>>客户端断开连接");
            try {
                socketChannel.close();
            } catch (IOException ex) {
                ex.printStackTrace();
            }
            selectionKey.cancel();
        }

    }


    /**
     * 广播给其他客户端
     *
     * @param selector
     */
    private void broadCast(Selector selector, SocketChannel sourceChannel, String request) {
        //获取所有已介入的客户端channel
        Set<SelectionKey> selectionKeySet = selector.keys();

        //循环向所有channel广播信息
        selectionKeySet.forEach(selectionKey -> {
            Channel targetChannel = selectionKey.channel();

            if (targetChannel instanceof SocketChannel && targetChannel != sourceChannel) {
                try {
                    ((SocketChannel) targetChannel).write(Charset.forName(CODING_TYPE).encode(request));
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }

        });
    }

    @Override
    public void run(ApplicationArguments args) {
//        log.info(".............sdadsadsada................");
//        while (true) {
//            try {
//                start();+
//            } catch (IOException e) {
//                try {
//                    selector.close();
//                    serverSocketChannel.close();
//                } catch (IOException ex) {
//                    ex.printStackTrace();
//                }
//                log.warn(">>>>>>>>>>>>>>>>>>共享纸巾服务端异常,正在重启<<<<<<<<<<<<<<<<<<<");
//            }
//        }
    }

}
